package com.shujia.flink.tf

import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

object Demo9SideOutPut {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val students: DataStream[String] = env.readTextFile("flink/data/students.txt")

    //spark 的方式
    //val nanDS: DataStream[String] = students.filter(_.split(",")(3) == "男")
    // val nvDS: DataStream[String] = students.filter(_.split(",")(3) == "女")


    val nan: OutputTag[String] = OutputTag[String]("男")
    val nv: OutputTag[String] = OutputTag[String]("女")


   val processDS: DataStream[String] =  students.process(new ProcessFunction[String, String] {
      override def processElement(value: String,
                                  ctx: ProcessFunction[String, String]#Context,
                                  out: Collector[String]): Unit = {

        //向下游发送数据，打上标记
        val gender: String = value.split(",")(3)

        if (gender.equals("男")) {
          ctx.output(nan, value)
        }else{
          ctx.output(nv, value)
        }

      }
    })

    //通过标记获取流
    val nanDS: DataStream[String] = processDS.getSideOutput(nan)

    val nvDS: DataStream[String] = processDS.getSideOutput(nv)


    nvDS.print()

    env.execute()

  }

}
